package com.bonjour.common.helper;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import okhttp3.Call;
import okhttp3.Callback;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.BufferedSource;

public final class OkSseHelper {
    private static final Logger LOG = LoggerFactory.getLogger(OkSseHelper.class);
    private OkSseHelper() {
    }

    public static ServerSentEvent subscribe(String url, Listener  listener) {
        ServerSentEvent sse = new ServerSentEvent(new Request.Builder().url(url).build(), listener);
        sse.connect(new OkHttpClient.Builder().readTimeout(0, TimeUnit.SECONDS).retryOnConnectionFailure(true).build());
        return sse;
    }

    public static class ServerSentEvent {
        private long reconnectTime = TimeUnit.SECONDS.toMillis(3);
        private final Request originalRequest;
        private final Listener listener;
        private OkHttpClient client;
        private Call call;
        private String lastEventId;
        ServerSentEvent(Request request, Listener listener) {
            if (!"GET".equals(request.method())) {
                throw new IllegalArgumentException("Request must be GET: " + request.method());
            }
            this.originalRequest = request;
            this.listener = listener;
        }
        void connect(OkHttpClient client) {
            this.client = client;
            prepareCall(originalRequest);
            enqueue();
        }
        private void prepareCall(Request request) {
            if (client == null) {
                throw new AssertionError("Client is null");
            }
            Request.Builder requestBuilder = request.newBuilder()
                                                    .header("Accept-Encoding", "")
                                                    .header("Accept", "text/event-stream")
                                                    .header("Cache-Control", "no-cache");

            if (lastEventId != null) {
                requestBuilder.header("Last-Event-Id", lastEventId);
            }

            call = client.newCall(requestBuilder.build());
        }
        private void enqueue() {
            //noinspection NullableProblems
            call.enqueue(new Callback() {
                @Override
                public void onFailure(Call call, IOException e) {
                    notifyFailure(e);
                }
                @Override
                public void onResponse(Call call, Response response) {
                    if (response.isSuccessful()) {
                        openSse(response);
                    } else {
                        notifyFailure(new IOException(response.message()));
                    }
                }
            });
        }
        private void openSse(Response response) {
            try (ResponseBody body = response.body()) {
                if (null == body){
                    return;
                }
                Reader sseReader = new Reader(body.source());
                sseReader.setTimeout();
                listener.onOpen(this, response);
                //noinspection StatementWithEmptyBody
                while (call != null && !call.isCanceled() && sseReader.read()) {
                }
            }
        }
        private void notifyFailure(Throwable throwable) {
            if (!retry(throwable)) {
                listener.onClosed(this);
                close();
            }
        }
        private boolean retry(Throwable throwable) {
            if (!Thread.currentThread().isInterrupted() && !call.isCanceled() && listener.onRetry(this, throwable)) {
                prepareCall(originalRequest);
                try {
                    Thread.sleep(reconnectTime);
                } catch (InterruptedException ignored) {
                    return false;
                }
                if (!Thread.currentThread().isInterrupted() && !call.isCanceled()) {
                    enqueue();
                    return true;
                }
            }
            return false;
        }
        public Request request() {
            return originalRequest;
        }
        public void close() {
            if (call != null && !call.isCanceled()) {
                call.cancel();
            }
        }

        private class Reader {
            private static final char COLON_DIVIDER = ':';
            private static final String DATA = "data";
            private static final String ID = "id";
            private static final String EVENT = "event";
            private static final String RETRY = "retry";
            private static final String DEFAULT_EVENT = "message";
            private static final String EMPTY_STRING = "";
            private final Pattern DIGITS_ONLY = Pattern.compile("^[\\d]+$");
            private final BufferedSource source;
            private StringBuilder data = new StringBuilder();
            private String eventName = DEFAULT_EVENT;

            Reader(BufferedSource source) {
                this.source = source;
            }
            boolean read() {
                try {
                    String line = source.readUtf8LineStrict();
                    processLine(line);
                } catch (IOException e) {
                    notifyFailure(e);
                    return false;
                }
                return true;
            }

            void setTimeout() {
                if (source != null) {
                    source.timeout().timeout(0, TimeUnit.MILLISECONDS);
                }
            }
            private void processLine(String line) {
                if (line == null || line.isEmpty()) {
                    dispatchEvent();
                    return;
                }
                int colonIndex = line.indexOf(COLON_DIVIDER);
                // If line starts with COLON dispatch a comment
                if (colonIndex == 0) {
                    listener.onComment(ServerSentEvent.this, line.substring(1).trim());
                }
                // Collect the characters on the line after the first U+003A COLON character (:), and let value be that string.
                else if (colonIndex != -1) {
                    String field = line.substring(0, colonIndex);
                    String value = EMPTY_STRING;
                    int valueIndex = colonIndex + 1;
                    if (valueIndex < line.length()) {
                        // If value starts with a single U+0020 SPACE character, remove it from value.
                        if (line.charAt(valueIndex) == ' ') {
                            valueIndex++;
                        }
                        value = line.substring(valueIndex);
                    }
                    processField(field, value);
                } else {
                    processField(line, EMPTY_STRING);
                }
            }

            private void dispatchEvent() {
                if (data.length() == 0) {
                    return;
                }
                String dataString = data.toString();
                if (dataString.endsWith("\n")) {
                    dataString = dataString.substring(0, dataString.length() - 1);
                }
                listener.onMessage(ServerSentEvent.this, lastEventId, eventName, dataString);
                data.setLength(0);
                eventName = DEFAULT_EVENT;
            }
            private void processField(String field, String value) {
                if (DATA.equals(field)) {
                    data.append(value).append('\n');
                } else if (ID.equals(field)) {
                    lastEventId = value;
                } else if (EVENT.equals(field)) {
                    eventName = value;
                } else if (RETRY.equals(field) && DIGITS_ONLY.matcher(value).matches()) {
                    long timeout = Long.parseLong(value);
                    reconnectTime = timeout > 0 ? timeout : reconnectTime;
                }
            }
        }
    }

    public interface Listener {
        default void onOpen(@SuppressWarnings("unused") ServerSentEvent sse, Response response){
            LOG.info("SSE open response={}", response.toString());
        }
        default void onComment(@SuppressWarnings("unused") ServerSentEvent sse, String comment){
            LOG.info("SSE on comment={}", comment);
        }
        void onMessage(ServerSentEvent sse, String id, String event, String message);
        boolean onRetry(ServerSentEvent sse, Throwable error);
        void onClosed(ServerSentEvent sse);
    }
}
